package cn.bblocks.cacheTest.test;

/**
* @Description:TODO
* @author 作者:jzhao
* @createDate 创建时间：2020年8月29日 上午1:29:00
* @version 1.0     
*/
import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class WordCount {

	public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {

		private final static IntWritable one = new IntWritable(1); // 词的个数，刚开始都为1，也可以不定义，直接context.write(keytext, 1);
		private Text word = new Text(); // 定义一个text对象，用来充当中间变量，存储词，
		// Text类型相当于java中的String类型，IntWritable相当于java中的Integer类型

		public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
			String str = value.toString(); // 获取值
			StringTokenizer itr = new StringTokenizer(str); // StringTokenizer根据空格等分隔字符串到stringTokenizer
			while (itr.hasMoreTokens()) { // 返回是否还有分隔符，判断是否还有单词
				word.set(itr.nextToken()); // nextToken()：返回从当前位置到下一个分隔符的字符串。
				context.write(word, one); // context.write("hello",1)
				// System.out.println("qiyadeng map message:"+word+"/"+one);

			}
		}
	}

	public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> { // 前两个输入：例：（hello，1），后两个输出（hello，2）
		private IntWritable result = new IntWritable();

		public void reduce(Text key, Iterable<IntWritable> values, Context context // 它用来与MapReduce系统进行通信，如把map的结果传给reduce处理
		) throws IOException, InterruptedException {
			int sum = 0;

			for (IntWritable val : values) {
				sum += val.get();
			}
			result.set(sum); // 如 对于hello，sum是2
			context.write(key, result); // hello，2
			// System.out.println("qiyadeng reduce message:"+key+"/"+result);
		}
	}

	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
		if (otherArgs.length != 2) {
			System.err.println("Usage: wordcount <in> <out>");
			System.exit(2);
		}
		Job job = new Job(conf, "word count"); // job对象指定作业执行规范
		job.setJarByClass(WordCount.class); // 代码打包成一个jar文件（hadoop在集群上发布这个文件）
		job.setMapperClass(TokenizerMapper.class); // 指定map类型
		job.setCombinerClass(IntSumReducer.class); //
		job.setReducerClass(IntSumReducer.class); // 指定reduce类型
		job.setOutputKeyClass(Text.class); // 控制map函数的输出类型
		job.setOutputValueClass(IntWritable.class); // 控制reduce函数的输出类型
		// 构造job对象之后，指定输入输出数据的路径
		FileInputFormat.addInputPath(job, new Path(otherArgs[0])); // 定义输入数据的路径，该方法可以多次调用
		FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); // 指定输出路径，只能有一个输出路径，该方法指定的是
		// reduce函数输出文件的写入目录，在运行作业前该目录是不存在的
		System.exit(job.waitForCompletion(true) ? 0 : 1); // job.waitForCompletion(）方法提交作业并等待执行完成
	}
}
